package com.kele.search.service;

import com.kele.model.goods.SearchGoods;
import com.kele.search.feign.GoodsFeignClient;
import com.kele.search.mapper.ESSyncMapper;
import com.kele.search.model.ESSync;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class SyncService {

    @Autowired
    RestHighLevelClient restHighLevelClient;

    @Autowired
    ESSyncMapper esSyncMapper;

    @Autowired
    GoodsFeignClient goodsFeignClient;

    @Autowired
    RabbitTemplate rabbitTemplate;

    //@Scheduled(cron = "0/15 * * * * ? ")
    @Scheduled(cron = "0/15 * * * * ? ")
    private void db2ES() throws IOException {
        //1.比如 已经到8:02
        System.out.println("开始同步" + new Date());

        //2.获取ES 上次同步的时间,比如 最后同步的时间8:00
        ESSync esSync = esSyncMapper.selectById(1);
        Date lastSyncTime = esSync.getLastSyncTime();

        //3.获取自从上次同步完后修改的数据,8:00之后修改的数据
        //如果lastSyncTime ==null ，全量同步
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String format = dateFormat.format(lastSyncTime);
        List<SearchGoods> list = goodsFeignClient.findUpdatedAfter(format);

        //4.设置 最后同步时间
        esSync.setLastSyncTime(new Date());

        //5 同步
        if(list!=null && list.size()>0) {
            BulkRequest bulkRequest = new BulkRequest();
            for (SearchGoods searchGoods : list) {
                //1新增文档 构建IndexRequest对象
                IndexRequest indexRequest = new IndexRequest("kele_goods");
                //2.封装请求对象
                Map<String, Object> goods1 = new HashMap<>();
                goods1.put("goodsid", searchGoods.getGoodsId());
                goods1.put("brand", searchGoods.getBrand());
                goods1.put("category", searchGoods.getCategory());
                goods1.put("title", searchGoods.getTitle());
                goods1.put("price", searchGoods.getPrice());
                goods1.put("update_time", searchGoods.getUpdateTime());
                goods1.put("main_pic", searchGoods.getMainPic());
                //执行文档插入操作
                indexRequest.id(searchGoods.getGoodsId().toString());

                indexRequest.source(goods1);
                //IndexResponse indexRes = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
                //3.获取响应结果
                //System.out.println(indexRes.status());
                //System.out.println(indexRes.toString());
                bulkRequest.add(indexRequest);

                rabbitTemplate.convertAndSend("page_exchange_fanout","",searchGoods.getGoodsId());

            }
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            System.out.println(bulk.status());
            System.out.println(bulk.toString());
            //6.保存最后同步的时间
            esSyncMapper.updateById(esSync);//
        }

    }
}
